package org.teiid.translator.cassandra.execution;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;

import org.teiid.language.Select;
import org.teiid.logging.LogConstants;
import org.teiid.logging.LogManager;
import org.teiid.translator.DataNotAvailableException;
import org.teiid.translator.ExecutionContext;
import org.teiid.translator.ResultSetExecution;
import org.teiid.translator.TranslatorException;
import org.teiid.translator.cassandra.CassandraConnection;
import org.teiid.translator.cassandra.CassandraExecutionFactory;

import com.datastax.driver.core.Row;

/**
 * 
 * @author Kouba
 *
 */
public class CassandraQueryExecution implements ResultSetExecution {
	
	private List<Object> values = new ArrayList<Object>();
	private Select query;
	private CassandraConnection connection;
	private Iterator<Row> iterator;
	private ExecutionContext executionContext;
	
	public CassandraQueryExecution(Select query, CassandraConnection connection, ExecutionContext context){
		this.query = query;
		this.connection = connection;
		this.executionContext = context;
	}

	@Override
	public void close() {
		LogManager.logDetail(LogConstants.CTX_CONNECTOR, CassandraExecutionFactory.UTIL.getString("close_query")); //$NON-NLS-1$
		
	}

	@Override
	public void cancel() throws TranslatorException {
		LogManager.logDetail(LogConstants.CTX_CONNECTOR, CassandraExecutionFactory.UTIL.getString("cancel_query")); //$NON-NLS-1$
		
	}

	@Override
	public void execute() throws TranslatorException {
		CassandraSQLVisitor visitor = new CassandraSQLVisitor();
		visitor.translateSQL(query);
		iterator = connection.executeQuery(visitor.getTranslatedSQL()).iterator();
	}

	@Override
	public List<?> next() throws TranslatorException, DataNotAvailableException {
		if (iterator != null && iterator.hasNext()){
			return getRow(iterator.next());
		}
		return null;
	}
	
	/**
	 * Returns {@code row} as list of objects representing values of columns in this row
	 * 
	 * @param row the row returned by iterator
	 * @return list of values in {@code row}
	 */
	private List<Object> getRow(Row row) {
		values.clear();
		for(int i = 0; i < row.getColumnDefinitions().size(); i++){
			switch(row.getColumnDefinitions().getType(i).getName()){
			case ASCII:
				values.add(row.getString(i));
				break;
			case BIGINT:
				values.add(row.getLong(i));
				break;
			case BLOB:
				values.add(row.getBytes(i));
				break;
			case BOOLEAN:
				values.add(row.getBool(i));
				break;
			case COUNTER:
				values.add(row.getLong(i));
				break;
			case DECIMAL:
				values.add(row.getDecimal(i));
				break;
			case DOUBLE:
				values.add(row.getDouble(i));
				break;
			case FLOAT:
				values.add(row.getFloat(i));
				break;
			case INET:
				values.add(row.getInet(i));
				break;
			case INT:
				values.add(row.getInt(i));
				break;
			case LIST:
				values.add(row.getList(i, row.getColumnDefinitions().getType(i).getTypeArguments().get(0).asJavaClass()));
				break;
			case MAP:
				values.add(row.getMap(i, row.getColumnDefinitions().getType(i).getTypeArguments().get(0).asJavaClass(),
										 row.getColumnDefinitions().getType(i).getTypeArguments().get(1).asJavaClass()));
				break;
			case SET:
				values.add(row.getSet(i, row.getColumnDefinitions().getType(i).getTypeArguments().get(0).asJavaClass()));
				break;
			case TEXT:
				values.add(row.getString(i));
				break;
			case TIMESTAMP:
				values.add(row.getDate(i));
				break;
			case TIMEUUID:
				values.add(row.getUUID(i));
				break;
			case UUID:
				values.add(row.getUUID(i));
				break;
			case VARCHAR:
				values.add(row.getString(i));
				break;
			case VARINT:
				values.add(row.getVarint(i));
				break;
			default:
				values.add(row.getBytesUnsafe(i));
				break;
			}
			
		}
		return values;
	}

}
